resolve TODO

find a better name and make it shorter

encapsulate in struct

comment

rename to addresses

think about way not to clone the addresses

use structs for complex types

dont use asserts in prod code

Use index 0 instead of poping and unwrapping

use index 0 for storing aggregated_response

Lets try to use pipeline for both transaction and pipeline, differentiating by is_atomic

lets remove Transaction and use Pipeline + is_atomic flag

Please also add comments there describing things there

From protocol's view the only difference between pipeline and transaction is 2 extra commands MULTI and EXEC

also being atomic + multi slots enabled

Please add documentation to the function and state that for non atomic pipelines it returns none

1. Don't use unwrap in production code. If a bug would cause pipeline_map.keys() to be empty it would crash the whole client. Instead, change this function to return a result and return ClientError if no random address is found.
2. Coping all addresses is redundant, you can achieve the same with:

    let mut rng = rand::thread_rng();
    if let Some(node_context) = pipeline_map
        .values_mut()
        .choose(&mut rng) {
            node_context.add_command(cmd, index, None);
            Ok(())
    } else {
        // return error
    }

The name of the function should declare that it's only relevant for pipelines, the current name is misleading

add use crate::cluster_routing::command_for_multi_slot_indices and remove the prefix

missing description, it ain't clear when and why it would be used

same issue - all of these functions are placed under ClusterConnInner though they are only relevant for pipelines. I'm not sure there is a good reason placing them here. Why not moving all of the pipelines logic into a separate file (e.g. pipeline, pipeline_routing) under the async_cluster folder?

why pub? pub refer to user-facing APIs

see if it can be removed or if pub(crate) is enough

same - function shouldn't be pub

I read the description 3 times and it's still unclear to me what this function does :(
maybe a simple example would help.

what do you mean by "replaces the existing entries in values_and_addresses for the given command index."? do you mean that it sorts the entries in values_and_addresses by the command indices calculated in this function? try to make it clearer

Please create type aliases for the return type: (Vec<(usize, Option)>, Vec, String), it isn't readable

we can make this complex type a bit clearer with type aliases:

// define on top
type NodeResponse = (Value, String); // A response with its source node address.
type PipelineResponses = Vec<Vec<NodeResponse>>; // Outer Vec: pipeline commands, Inner Vec: (response, address).
...

let mut values_and_addresses: PipelineResponses = vec![Vec::new(); pipeline.len()];

and it can also be used elsewhere (e.g. aggregate_pipeline_multi_node_commands)

same-type alias

then remove #[allow(clippy::type_complexity)]

since returning Ok with Some(err) is confusing, you can make it more readable with some enum representing the return values, something like

enum MultiPipelineResult {
    /// All tasks completed successfully.
    AllSuccessful,

    /// Some tasks failed, returning the first encountered error and the associated operation target.
    FirstError {
        target: OperationTarget,
        error: RedisError,
    },
}

what happens if conn is None?

it isn't an IO error.

oh ok I see now that you did this logic outside to this function. I think it would be more readable if youll add all of this logic into the handle_pipeline_single_node_routing function, then you can skip this redundant error handling by something like:

async fn handle_pipeline_single_node_routing(...) {
     if matches!(routing, InternalSingleNodeRouting::Random) && !pipeline_map.is_empty() {
        let mut rng = rand::thread_rng();
        if let Some(node_context) = pipeline_map.values_mut().choose(&mut rng) {
            node_context.add_command(cmd, index, None);
            return Ok(()); 
        }
    }

    let (address, conn) =
        ClusterConnInner::get_connection(routing, core, Some(Arc::new(cmd.clone())))
            .await
            .map_err(|err| (OperationTarget::NotFound, err))?;

    add_command_to_node_pipeline_map(pipeline_map, address, conn, cmd, index, None);
    Ok(())

nit: you can remove this line and line 156 and return a single Ok(()) at the end of the function

It isn't safe and we can't assume it, check that it isn't empty and raise an exception if it does

nit: when you hardcode types in the documentation it creates another place we'll have to remember to maintain if we'll change it.

"".to_string()?

why does it return an empty string instead of some representing enum/None

do you still need this field at this stage? can't you completely filter it out?

I cant filter it out since I am only going through the commands in the pipeline with multi-node routing, the pipeline_responses still contains the rest of the commands (with single node routing) that are with the type of (response, address)

nit: import cluster_routing::RoutingInfo and remove all redundant prefixes

cloning all commands is expansive, is it necessary?

yes 😞

why?

suggestion - can we wrap the commands with Arc? if it's too complex, please add it as a todo, it can be done in a seperate PR

please add description so it would be clear how this function is different from send_transaction

handle_pipeline_request -> handle_multi_slot_pipeline_request / handle_non_atomic_pipeline or something like that.
now the code is confusing -

if atomic {
try_pipeline_request ()
} else {
handle_pipeline_request()
}

if the pipeline.len == 1, aren't we better off by using the try_pipeline_request function without all of this additional logic?

not exactly, if pipelines_by_node ==1 we can avoide all this logic, but pipeline.len == 1 can still contain a multi-node command

why?

I think that if we don't have any primary connections it is ok to set the client with AllConnectionsUnavailable

just make the error message clearer if it's due to all connections or only primary ones